cortex/do_runner/executor.py (243-249)

Executing commands received from an LLM using shell=True without proper validation is a critical security risk. A compromised or malicious LLM response could lead to arbitrary command execution on the user's system. While terminal.py includes a _is_safe_fix_command check, that protection is missing here.

Before executing the command, you should validate it against a security policy. You could adapt the CommandValidator from cortex.ask or, even better, reuse the _is_safe_fix_command logic from cortex/do_runner/terminal.py to ensure only safe commands are executed.

cortex/do_runner/diagnosis_v2.py (1474-1476)

The execute_fix_commands function constructs a command by substituting variables directly into a command template using string.replace(). The variable values can originate from various sources, including the user's query or LLM output, and are not sanitized. This creates a command injection vulnerability. For example, if a variable contained a value like "; rm -rf /, it would be executed as part of the command when shell=True is used.

To prevent this, you must sanitize all variable values before substituting them into the command string. The standard and safest way to do this in Python is with shlex.quote().

            import shlex
            for var_name, value in resolved_variables.items():
                command = command.replace(f"{{{var_name}}}", shlex.quote(value))

cortex/do_runner/terminal.py (1547-1550)

The DANGEROUS_PATTERNS list is a great security measure. However, the patterns for blocking command piping to a shell are a bit too specific and can be bypassed. For example, curl ... | /bin/bash or curl ... | zsh would not be caught.

To make this more robust, consider using a regex pattern that blocks piping to common shells, regardless of their path, as recommended by the repository's general rules.

            r'|\s*(?:.*/)?(?:bash|sh|zsh)\b', # Block piping to common shells

cortex/cli.py (851-1129)

This function, _run_interactive_do_session, is quite long (nearly 300 lines) and handles many different responsibilities, including session setup, signal handling, UI rendering, and command processing logic for history, help, clear, etc. This reduces readability and makes the function difficult to maintain and test.

Consider refactoring this into several smaller, more focused helper methods. For example:

• _setup_interactive_session(): To handle the initial setup and welcome panel.
• _handle_session_command(query, ...): To process meta-commands like history, clear, and help.
• _process_user_query(query, handler, ...): For the main logic of passing the query to the AskHandler.
• _render_session_history(history): To encapsulate the rich.table logic for displaying history.

Breaking down this function will improve modularity and make the interactive session flow easier to follow.

cortex/system_info_generator.py-149-152 (1)

149-152: Template commands conflict with CommandValidator allowlist.

ufw status is blocked and aa-status is not in the allowed list, so these templates will always fail. Either update the validator to allow these read‑only commands or replace them with approved alternatives.

cortex/watch_service.py-68-73 (1)

68-73: Harden log file permissions to avoid leaking command history.

These logs can contain sensitive commands; with a typical umask they may be world‑readable. Ensure .cortex and its log files are restricted (e.g., dir 0700, files 0600).

🔒 Example fix (apply to all log files)

     def log(self, message: str):
         """Log a message to the service log."""
         log_file = self.cortex_dir / "watch_service.log"
         timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         with open(log_file, "a") as f:
             f.write(f"[{timestamp}] {message}\n")
+        os.chmod(log_file, 0o600)

Also applies to: 392-394, 434-435

cortex/do_runner/terminal.py-717-739 (1)

717-739: Don’t overwrite existing PROMPT_COMMAND.

The hook currently replaces any existing PROMPT_COMMAND, which can break user shells. Preserve and append instead, and avoid re‑adding if already present.

✅ Proposed fix

-    echo "${{tty_name:-unknown}}|$cmd" >> {watch_file}
-}}
-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+    echo "${{tty_name:-unknown}}|$cmd" >> {watch_file}
+}}
+if [[ -z "$PROMPT_COMMAND" ]]; then
+    export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+elif [[ "$PROMPT_COMMAND" != *"__cortex_log_cmd"* ]]; then
+    export PROMPT_COMMAND="${{PROMPT_COMMAND}}; __cortex_log_cmd"
+fi

cortex/watch_service.py-524-536 (1)

524-536: Ensure stop() only returns success if the daemon actually exited.

After waiting, the code returns success even if the PID still exists.

✅ Proposed fix

             for _ in range(10):
                 try:
                     os.kill(pid, 0)
                     time.sleep(0.5)
                 except ProcessLookupError:
                     break
-
-            return True, f"Service stopped (PID: {pid})"
+            try:
+                os.kill(pid, 0)
+                return False, f"Service still running (PID: {pid})"
+            except ProcessLookupError:
+                return True, f"Service stopped (PID: {pid})"

cortex/do_runner/terminal.py-1027-1056 (1)

1027-1056: Parse the new TTY|COMMAND log format.

When the watch service or hook writes tty|command, the monitor currently treats the entire line as the command, which pollutes command detection. Split and use the command part.

✅ Proposed fix

-                        # Handle format with timestamp: "HH:MM:SS command"
+                        # Handle format with terminal prefix: "TTY|COMMAND"
+                        if "|" in line and not re.match(r"^\d{2}:\d{2}:\d{2}\s+", line):
+                            parts = line.split("|", 1)
+                            if len(parts) == 2 and parts[1].strip():
+                                line = parts[1].strip()
+
+                        # Handle format with timestamp: "HH:MM:SS command"
                         if re.match(r"^\d{2}:\d{2}:\d{2}\s+", line):
                             parts = line.split(" ", 1)
                             if len(parts) == 2 and parts[1].strip():
                                 self._process_observed_command(parts[1].strip(), "live_terminal")
                         else:
                             # Plain command
                             self._process_observed_command(line, "live_terminal")

cortex/do_runner/terminal.py-2230-2284 (1)

2230-2284: Avoid shell=True with user‑derived tokens in verification commands.

service_name/container_name are derived from observed commands; building shell strings can execute unintended tokens (e.g., embedded ;). Use argv lists and trim output in Python.

✅ Safer pattern (example for systemctl)

-                check_cmd = f"systemctl status {service_name} 2>&1 | head -5"
+                safe_service = re.sub(r"[^a-zA-Z0-9_.@-]", "", service_name)
+                check_cmd = ["systemctl", "status", safe_service]
@@
-            try:
-                result = subprocess.run(
-                    check_cmd, shell=True, capture_output=True, text=True, timeout=5
-                )
-
-                output = result.stdout + result.stderr
+            try:
+                result = subprocess.run(
+                    check_cmd, capture_output=True, text=True, timeout=5
+                )
+                output = "\n".join((result.stdout + result.stderr).splitlines()[:5])

cortex/system_info_generator.py-327-362 (1)

327-362: Add missing return type hints for __init__ and _initialize_client.

Both methods lack return type annotations. Per coding guidelines requiring type hints on all Python code, add -> None: to these methods.

Proposed fix

     def __init__(
         self,
         api_key: str | None = None,
         provider: str = "claude",
         model: str | None = None,
         debug: bool = False,
-    ):
+    ) -> None:
         """
         Initialize the system info generator.

-    def _initialize_client(self):
+    def _initialize_client(self) -> None:
         """Initialize the LLM client."""

cortex/watch_service.py-30-98 (1)

30-98: Add explicit return type hints to all public and internal methods.

The module contains 15 methods lacking return type annotations, violating the project's typing requirement. Add return types to: __init__ (→ None), _handle_signal (→ None), _handle_reload (→ None), log (→ None), _load_state (→ None), _save_state (→ None), _monitor_bash_history (→ None), _monitor_watch_hook (→ None), _log_to_json (→ None), _log_command (→ None), _cleanup_stale_terminals (→ None), start (→ bool), _shutdown (→ None), stop (→ tuple[bool, str]), and main (→ None). Additionally, the signal handler methods are missing parameter type hints for signum and frame.

cortex/do_runner/models.py-90-97 (1)

90-97: Dead code: get_depth() contains an infinite loop and is never used.

The method assigns node = node (line 96), which will cause an infinite loop if parent_id is set. Additionally, codebase search shows this method is never called anywhere, indicating it is dead code.

The fundamental issue is that TaskNode has no reference to its parent node or to the containing TaskTree, making it impossible for this method to traverse up the tree correctly. Since TaskTree maintains a complete _all_tasks dictionary, the most practical solution is to remove this broken method from TaskNode entirely, or move the depth-calculation logic to TaskTree as a utility method if needed in the future.

Recommendation: Remove the unused method unless there is a planned feature that requires it. If depth calculation becomes necessary, implement it on TaskTree using the _all_tasks lookup mechanism.

cortex/do_runner/managers.py-220-240 (1)

220-240: Security: chmod fallback grants overly permissive access.

The _grant_privilege_chmod fallback uses o+rw (others read/write), which grants access to ALL users on the system, not just the cortex user. This is significantly more permissive than the intended setfacl behavior and creates a security risk.

Consider using group-based permissions as a safer fallback:

Suggested safer fallback

     `@classmethod`
     def _grant_privilege_chmod(cls, file_path: str, mode: str) -> tuple[bool, str]:
         """Fallback privilege granting using chmod."""
         try:
-            chmod_mode = ""
-            if "r" in mode:
-                chmod_mode = "o+r"
-            if "w" in mode:
-                chmod_mode = "o+rw" if chmod_mode else "o+w"
-            if "x" in mode:
-                chmod_mode = chmod_mode + "x" if chmod_mode else "o+x"
+            # Use group permissions instead of world-readable/writable
+            chmod_mode = "g+"
+            if "r" in mode:
+                chmod_mode += "r"
+            if "w" in mode:
+                chmod_mode += "w"
+            if "x" in mode:
+                chmod_mode += "x"
+            if chmod_mode == "g+":
+                chmod_mode = "g+r"  # Default to read

             subprocess.run(
-                ["sudo", "chmod", chmod_mode, file_path],
+                ["sudo", "chgrp", cls.CORTEX_GROUP, file_path],
+                check=True,
+                capture_output=True,
+            )
+            subprocess.run(
+                ["sudo", "chmod", chmod_mode, file_path],
                 check=True,
                 capture_output=True,
             )
-            return True, f"Granted {mode} access to {file_path} (chmod fallback)"
+            return True, f"Granted {mode} access to {file_path} (group chmod fallback)"

scripts/setup_ask_do.py-289-312 (1)

289-312: Preserve existing PROMPT_COMMAND instead of overwriting.

Both the hook file and .bashrc block replace PROMPT_COMMAND, which can break existing shell integrations (direnv, pyenv, starship, etc.). Compose with the existing value instead.

🛠️ Safer PROMPT_COMMAND composition

@@
-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+export PROMPT_COMMAND="history -a; __cortex_log_cmd${PROMPT_COMMAND:+; $PROMPT_COMMAND}"

@@
-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+export PROMPT_COMMAND="history -a; __cortex_log_cmd${{PROMPT_COMMAND:+; $PROMPT_COMMAND}}"

Also applies to: 328-347

scripts/setup_ask_do.py-43-68 (1)

43-68: Add explicit return type hints for public helpers and main().

The new helpers are public APIs in this module, but they omit return type hints (e.g., print_*, main). Please add -> None / -> int consistently. As per coding guidelines, type hints are required in Python code.

🧷 Example adjustments

-def print_header(text: str):
+def print_header(text: str) -> None:
@@
-def print_step(text: str):
+def print_step(text: str) -> None:
@@
-def print_success(text: str):
+def print_success(text: str) -> None:
@@
-def print_warning(text: str):
+def print_warning(text: str) -> None:
@@
-def print_error(text: str):
+def print_error(text: str) -> None:
@@
-def main():
+def main() -> int:

Also applies to: 544-562

scripts/setup_ask_do.py-70-87 (1)

70-87: Handle missing command binaries to avoid crashing the setup wizard.

run_cmd doesn’t catch FileNotFoundError, so calls with check=False (e.g., systemctl, cortex) can still raise and abort the script. Suggest handling this explicitly so the wizard degrades gracefully.

🧯 Suggested fix

@@
-    except subprocess.TimeoutExpired:
+    except FileNotFoundError:
+        print_error(f"Command not found: {' '.join(cmd)}")
+        if check:
+            raise
+        stdout = "" if capture else None
+        stderr = "command not found" if capture else None
+        return subprocess.CompletedProcess(cmd, 127, stdout=stdout, stderr=stderr)
+    except subprocess.TimeoutExpired:
         print_error(f"Command timed out: {' '.join(cmd)}")
         raise

cortex/do_runner/verification.py-193-208 (1)

193-208: Potential command injection via unsanitized container_name.

The container_name extracted from regex (line 190) is directly interpolated into shell commands (lines 194, 200, 206). If the input cmd contains crafted values like --name "foo; malicious_cmd", this could lead to command injection.

Consider validating/sanitizing extracted values or using parameterized execution where possible.

🛡️ Suggested mitigation

import shlex

# Validate container name format (alphanumeric, dashes, underscores)
if not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9_.-]*$', container_name):
    return result  # Invalid container name, skip check

# Use shlex.quote for shell safety
safe_name = shlex.quote(container_name)
success, container_id, _ = self._execute_command(
    f"docker ps -aq --filter name=^{safe_name}$", needs_sudo=False
)

cortex/do_runner/verification.py-23-34 (1)

23-34: Logic error: Double sudo execution when needs_sudo=True.

When needs_sudo=True, line 24-25 prepends "sudo " to cmd, then line 27-28 runs ["sudo", "bash", "-c", cmd]. This results in executing sudo bash -c "sudo <original_cmd>" — double sudo invocation.

🐛 Proposed fix

     def _execute_command(
         self, cmd: str, needs_sudo: bool = False, timeout: int = 120
     ) -> tuple[bool, str, str]:
         """Execute a single command."""
         try:
-            if needs_sudo and not cmd.strip().startswith("sudo"):
-                cmd = f"sudo {cmd}"
-
-            result = subprocess.run(
-                ["sudo", "bash", "-c", cmd] if needs_sudo else cmd,
-                shell=not needs_sudo,
-                capture_output=True,
-                text=True,
-                timeout=timeout,
-            )
+            if needs_sudo:
+                result = subprocess.run(
+                    ["sudo", "bash", "-c", cmd],
+                    capture_output=True,
+                    text=True,
+                    timeout=timeout,
+                )
+            else:
+                result = subprocess.run(
+                    cmd,
+                    shell=True,
+                    capture_output=True,
+                    text=True,
+                    timeout=timeout,
+                )
             return result.returncode == 0, result.stdout.strip(), result.stderr.strip()

cortex/do_runner/verification.py-602-607 (1)

602-607: Destructive default action: fuser -k kills processes without confirmation.

The stop_existing alternative action uses sudo fuser -k {port}/tcp which forcefully terminates processes. This could disrupt critical services. Consider using a less aggressive approach or adding a warning.

💡 Suggested safer alternative

                         {
                             "action": "stop_existing",
-                            "description": f"Stop process using port {port}",
-                            "commands": [f"sudo fuser -k {port}/tcp"],
+                            "description": f"Stop process using port {port} (WARNING: may disrupt services)",
+                            "commands": [
+                                f"sudo lsof -i :{port} -t",  # First identify the process
+                                f"# Review above PID, then: sudo kill <PID>",
+                            ],
                         },

cortex/do_runner/diagnosis_v2.py-1462-1471 (1)

1462-1471: Security concern: Password exposed in pip config and process arguments.

Line 1463 embeds credentials in a URL that gets written to pip config (visible in plaintext). Line 1470-1471 substitutes password into a command string executed via shell, which could expose it in process listings.

🔒 Suggested safer approach for pip

             elif "pip" in command.lower() or "pypi" in host.lower():
-                login_cmd = f"pip config set global.index-url https://{username}:{{password}}@pypi.org/simple/"
+                # Use environment variable or keyring instead of embedding in config
+                console.print("[yellow]For PyPI, consider using keyring or environment variables:[/yellow]")
+                console.print("  export TWINE_USERNAME=<username>")
+                console.print("  export TWINE_PASSWORD=<password>")
+                login_cmd = None  # Don't store password in config

cortex/do_runner/verification.py-1044-1053 (1)

1044-1053: Bug: operation_type is overwritten in loop, only last value used.

The operation_type variable is reassigned on each iteration (line 1052), so after the loop, only the last matched operation type is retained. This affects the logic at line 1102 where operation_type is checked for all target_files.

🐛 Proposed fix: Track operation type per file

-        target_files = []
-        operation_type = None
+        target_files: list[tuple[str, str]] = []  # (path, operation_type)

         for pattern, op_type in file_creation_patterns:
             matches = re.findall(pattern, cmd)
             for match in matches:
                 if match.startswith("/") or match.startswith("~"):
-                    target_files.append(match)
-                    operation_type = op_type
+                    target_files.append((match, op_type))

-        result["files_checked"] = target_files
+        result["files_checked"] = [f[0] for f in target_files]

-        for file_path in target_files:
+        for file_path, operation_type in target_files:

cortex/do_runner/Untitled-1-1 (1)

1-1: Remove accidental placeholder file.

This file contains only a stray character and should be deleted to avoid shipping meaningless artifacts.

🧹 Proposed fix

-i

cortex/demo.py-49-55 (1)

49-55: Remove trailing whitespace (Ruff W291) on Line 54.

This currently fails linting.

🧹 Proposed fix

-• **Install** - Install packages with AI interpretation  
+• **Install** - Install packages with AI interpretation

cortex/do_runner/terminal.py-185-305 (1)

185-305: Add explicit return type hints for public methods.

The public methods __init__ and start in the specified range (lines 185-305) are missing return type annotations. Additionally, start_monitoring (line 319) lacks a return type hint. All three should be annotated with -> None for consistency with coding guidelines requiring type hints on all public APIs.

scripts/setup_ask_do.sh-257-260 (1)

257-260: Remove unused variable CORTEX_PATH.

The static analysis correctly identified that CORTEX_PATH is assigned but never used. This should be removed to avoid confusion.

Suggested fix

         # Get Python path
         PYTHON_PATH=$(which python3)
-        CORTEX_PATH=$(which cortex 2>/dev/null || echo "$HOME/.local/bin/cortex")

cortex/do_runner/models.py-267-269 (1)

267-269: Minor: Command truncation always appends "..." even for short commands.

The print output unconditionally appends ... after truncating to 50 characters, which will display as short_cmd... even when the command is shorter than 50 characters.

Suggested fix

-        console.print(
-            f"{indent}{prefix}{icon} [{color}][{node.task_type.value}][/{color}] {node.command[:50]}..."
-        )
+        cmd_display = node.command[:50] + "..." if len(node.command) > 50 else node.command
+        console.print(
+            f"{indent}{prefix}{icon} [{color}][{node.task_type.value}][/{color}] {cmd_display}"
+        )

cortex/do_runner/managers.py-259-270 (1)

259-270: chmod revoke fallback may remove unrelated permissions.

Similar to the grant fallback, chmod o-rwx removes permissions for ALL "others", not just the cortex user. This could unintentionally break access for other legitimate users or services.

cortex/do_runner/database.py-129-157 (1)

129-157: Fix trailing whitespace flagged by linter.

Static analysis reports trailing whitespace on lines 130, 148, 149, and 150. Remove the trailing spaces to pass the lint check.

Fix trailing whitespace

         cursor = conn.execute("""
-            SELECT run_id, full_data FROM do_runs 
+            SELECT run_id, full_data FROM do_runs
             WHERE total_commands IS NULL OR total_commands = 0 OR commands_list IS NULL
         """)
         # ... lines 146-156 ...
                 conn.execute(
                     """
-                    UPDATE do_runs SET 
-                        total_commands = ?, 
-                        successful_commands = ?, 
+                    UPDATE do_runs SET
+                        total_commands = ?,
+                        successful_commands = ?,
                         failed_commands = ?,

cortex/do_runner/database.py-276-298 (1)

276-298: Missing null check for full_data before JSON parsing.

If full_data is NULL in the database, json.loads(row["full_data"]) at line 284 will raise TypeError. Add defensive handling.

Suggested fix

             if row:
+                if not row["full_data"]:
+                    return None
                 full_data = json.loads(row["full_data"])

cortex/do_runner/managers.py-56-56 (1)

56-56: Class attribute USER_PROTECTED_PATHS is shared across all instances.

USER_PROTECTED_PATHS is defined as a class attribute (set[str] = set()), meaning it's shared across all instances of ProtectedPathsManager. The _load_user_paths method assigns to self.USER_PROTECTED_PATHS, which shadows the class attribute. This could lead to unexpected behavior if multiple instances are created.

Suggested fix

 class ProtectedPathsManager:
     """Manages the list of protected files and folders requiring user authentication."""

     SYSTEM_PROTECTED_PATHS: set[str] = {
         # ... system paths ...
     }

-    USER_PROTECTED_PATHS: set[str] = set()
-
     def __init__(self):
+        self.user_protected_paths: set[str] = set()
         self.config_file = Path.home() / ".cortex" / "protected_paths.json"
         self._ensure_config_dir()
         self._load_user_paths()

Then update all references from self.USER_PROTECTED_PATHS to self.user_protected_paths.

cortex/cli.py-3972-3975 (1)

3972-3975: Truncate run output to avoid terminal flooding.

cmd["output"]/cmd["error"] can be large; the fallback path already truncates. Consider a small cap here too.

✂️ Suggested truncation

@@
-                    if cmd["output"]:
-                        console.print(f"   [dim]Output:[/dim] {cmd['output']}")
-                    if cmd["error"]:
-                        console.print(f"   [red]Error:[/red] {cmd['error']}")
+                    if cmd["output"]:
+                        output = cmd["output"]
+                        output = output[:250] + "..." if len(output) > 250 else output
+                        console.print(f"   [dim]Output:[/dim] {output}")
+                    if cmd["error"]:
+                        err = cmd["error"]
+                        err = err[:250] + "..." if len(err) > 250 else err
+                        console.print(f"   [red]Error:[/red] {err}")

cortex/do_runner/verification.py-302-305 (1)

302-305: Incomplete filtering for system paths.

The check if path in ["/dev/null", "/etc/os-release", "/proc/", "/sys/"] uses exact match, but paths like /proc/cpuinfo or /sys/class/net won't match. Use startswith() for directory prefixes.

🐛 Proposed fix

         for path in paths_in_cmd:
             # Skip common read paths
-            if path in ["/dev/null", "/etc/os-release", "/proc/", "/sys/"]:
+            if path == "/dev/null" or path == "/etc/os-release" or path.startswith("/proc/") or path.startswith("/sys/"):
                 continue

cortex/demo.py (1)

16-24: Add a timeout to subprocess calls to avoid hanging the demo.

If cortex blocks, the demo becomes unresponsive; a reasonable timeout makes it safer.

⏱️ Example adjustment

-        result = subprocess.run(cmd, capture_output=True, text=True)
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
@@
-        result = subprocess.run(cmd)
+        result = subprocess.run(cmd, timeout=60)

docs/ASK_DO_ARCHITECTURE.md (1)

451-463: Consider adding language specifiers to log format examples.

The static analysis tool flagged missing language specifiers on fenced code blocks. While ASCII diagrams don't benefit from syntax highlighting, the log file format examples could use text or log specifiers for consistency.

Suggested fix

-```
+```text
 pts_1|docker ps
 pts_1|sudo systemctl restart nginx

-```json
+```json
 {"timestamp": "2026-01-16T14:15:00.123", ...}

scripts/setup_ask_do.sh (1)

303-356: Code duplication: Watch hook logic defined twice.

The watch hook function __cortex_log_cmd is duplicated—once in watch_hook.sh (lines 303-325) and again inline in .bashrc (lines 334-356). This duplication could lead to maintenance issues if the logic needs updating.

Consider having .bashrc source the hook file instead of duplicating the code:

Suggested approach

 # Add to .bashrc
 MARKER="# Cortex Terminal Watch Hook"
 if [ -f ~/.bashrc ]; then
     if ! grep -q "$MARKER" ~/.bashrc; then
         print_step "Adding hook to .bashrc..."
         cat >> ~/.bashrc << 'EOF'

 # Cortex Terminal Watch Hook
-__cortex_last_histnum=""
-__cortex_log_cmd() {
-    # ... duplicated code ...
-}
-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+if [ -f "$HOME/.cortex/watch_hook.sh" ]; then
+    source "$HOME/.cortex/watch_hook.sh"
+fi

 alias cw="source ~/.cortex/watch_hook.sh"
 EOF

cortex/do_runner/executor.py (4)

37-49: Type hint for user_manager parameter is incorrect.

The user_manager parameter is typed as type (a metaclass), but based on usage it should be typed as the actual manager class or instance. Looking at the code, this appears to be passed but never used in this file.

Suggested fix

     def __init__(
         self,
-        user_manager: type,
+        user_manager: Any,
         paths_manager: Any,
         llm_callback: Callable[[str], dict] | None = None,
     ):

Alternatively, if CortexUserManager is the expected type, import and use it directly for better type safety.

---

330-419: Move repeated import re to module level.

The re module is imported three times inside conditional branches (lines 341, 367, 394). This is inefficient and unconventional. Move the import to the top of the file with other imports.

Suggested fix

Add at top of file:

import re

Then remove the inline imports at lines 341, 367, and 394.

---

54-64: Consider validating command entries before adding to tree.

Empty or malformed command entries (missing "command" key) will result in empty strings being added to the task tree. Consider adding validation.

Suggested improvement

     def build_tree_from_commands(
         self,
         commands: list[dict[str, str]],
     ) -> TaskTree:
         """Build a task tree from a list of commands."""
         for cmd in commands:
-            self.tree.add_root_task(
-                command=cmd.get("command", ""),
-                purpose=cmd.get("purpose", ""),
-            )
+            command = cmd.get("command", "").strip()
+            if command:  # Skip empty commands
+                self.tree.add_root_task(
+                    command=command,
+                    purpose=cmd.get("purpose", ""),
+                )
         return self.tree

---

114-246: Dead code: was_repaired variable is unused.

The was_repaired variable is initialized at line 121 but is never set to True. When repairs succeed, the method returns (True, True) directly (lines 192, 221, 241), and the final return at line 246 always returns (False, was_repaired) which is (False, False).

Additionally, lines 169-192 and 194-221 contain duplicated repair execution logic that could be refactored.

Suggested fix for dead variable

Either remove was_repaired entirely and simplify the returns, or use it properly:

-        was_repaired = False
         # ... existing code ...
-        return False, was_repaired
+        return False, False

cortex/do_runner/managers.py (1)

104-116: Minor redundancy in is_protected logic.

Line 113 checks path == protected, but this case is already handled by the exact match check at line 109. The redundant check can be removed.

Simplified version

     def is_protected(self, path: str) -> bool:
         """Check if a path requires authentication for access."""
         path = os.path.abspath(path)
         all_protected = self.SYSTEM_PROTECTED_PATHS | self.USER_PROTECTED_PATHS

         if path in all_protected:
             return True

         for protected in all_protected:
-            if path.startswith(protected + "/") or path == protected:
+            if path.startswith(protected + "/"):
                 return True

         return False

cortex/do_runner/database.py (3)

122-127: Silent exception handling may hide migration errors.

The except sqlite3.OperationalError: pass at lines 126-127 silently ignores all operational errors during schema migration. While this handles the case where a column already exists, it also hides legitimate errors (e.g., disk full, database corruption).

Consider more specific handling

             if col_name not in existing_columns:
                 try:
                     conn.execute(f"ALTER TABLE do_runs ADD COLUMN {col_name} {col_type}")
-                except sqlite3.OperationalError:
-                    pass
+                except sqlite3.OperationalError as e:
+                    # Column likely already exists from a previous migration
+                    if "duplicate column" not in str(e).lower():
+                        console.print(f"[yellow]Warning: Schema migration issue: {e}[/yellow]")

---

404-416: Session ID generation has lower entropy than run ID generation.

create_session uses md5 with a predictable timestamp, while _generate_run_id uses sha256 with os.urandom(16). For consistency and better uniqueness guarantees, consider using the same approach.

Suggested improvement

     def create_session(self) -> str:
         """Create a new session and return the session ID."""
-        session_id = f"session_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}_{hashlib.md5(str(datetime.datetime.now().timestamp()).encode()).hexdigest()[:8]}"
+        timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S%f")
+        random_part = hashlib.sha256(os.urandom(16)).hexdigest()[:8]
+        session_id = f"session_{timestamp}_{random_part}"

---

378-402: Add defensive handling for corrupted full_data in bulk retrieval.

get_recent_runs iterates rows and parses full_data without exception handling. A single corrupted record would cause the entire method to fail. Consider wrapping in try-except to skip malformed entries.

Suggested improvement

             runs = []
             for row in cursor:
-                full_data = json.loads(row["full_data"])
-                run = DoRun(
-                    run_id=full_data["run_id"],
-                    # ...
-                )
-                run.session_id = row["session_id"]
-                runs.append(run)
+                try:
+                    if not row["full_data"]:
+                        continue
+                    full_data = json.loads(row["full_data"])
+                    run = DoRun(
+                        run_id=full_data["run_id"],
+                        summary=full_data["summary"],
+                        mode=RunMode(full_data["mode"]),
+                        commands=[CommandLog.from_dict(c) for c in full_data["commands"]],
+                        started_at=full_data.get("started_at", ""),
+                        completed_at=full_data.get("completed_at", ""),
+                        user_query=full_data.get("user_query", ""),
+                        files_accessed=full_data.get("files_accessed", []),
+                        privileges_granted=full_data.get("privileges_granted", []),
+                    )
+                    run.session_id = row["session_id"]
+                    runs.append(run)
+                except (json.JSONDecodeError, KeyError, TypeError):
+                    continue  # Skip malformed records
             return runs

cortex/do_runner/verification.py (3)

19-38: Duplicate _execute_command method across three classes.

The _execute_command method is identically implemented in ConflictDetector, VerificationRunner, and FileUsefulnessAnalyzer. This violates DRY and creates maintenance burden.

♻️ Suggested refactor: Extract to a base class or helper module

+class CommandExecutor:
+    """Base class providing command execution capabilities."""
+
+    def _execute_command(
+        self, cmd: str, needs_sudo: bool = False, timeout: int = 120
+    ) -> tuple[bool, str, str]:
+        """Execute a single command."""
+        try:
+            if needs_sudo and not cmd.strip().startswith("sudo"):
+                cmd = f"sudo {cmd}"
+
+            result = subprocess.run(
+                ["sudo", "bash", "-c", cmd] if needs_sudo else cmd,
+                shell=not needs_sudo,
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+            )
+            return result.returncode == 0, result.stdout.strip(), result.stderr.strip()
+        except subprocess.TimeoutExpired:
+            return False, "", f"Command timed out after {timeout} seconds"
+        except Exception as e:
+            return False, "", str(e)
+
+
-class ConflictDetector:
+class ConflictDetector(CommandExecutor):
    """Detects conflicts with existing configurations."""
-
-    def _execute_command(
-        self, cmd: str, needs_sudo: bool = False, timeout: int = 120
-    ) -> tuple[bool, str, str]:
-        ...

Also applies to: 817-836, 997-1016

---

932-950: Arbitrary limit of 5 files may miss important verification targets.

The [:5] slice on line 932 limits file existence checks to 5 files. This arbitrary limit could skip verification of important configuration files. Consider making this configurable or documenting the rationale.

+    MAX_FILE_CHECKS = 5  # Class constant, configurable
+
     # File existence tests
-    for file_path in list(files_to_check)[:5]:
+    for file_path in list(files_to_check)[:self.MAX_FILE_CHECKS]:

---

1222-1231: Magic number for keyword overlap threshold.

The threshold > 2 on line 1227 is a magic number. Consider extracting it as a class constant for clarity and configurability.

+    MIN_KEYWORD_OVERLAP = 2  # Minimum keyword matches to consider content related
+
     # Generic - check for keyword overlap
     else:
         purpose_words = set(purpose_lower.split())
         content_words = set(content_lower.split())
         overlap = purpose_words & content_words

-        if len(overlap) > 2:
+        if len(overlap) > self.MIN_KEYWORD_OVERLAP:

cortex/do_runner/diagnosis_v2.py (6)

20-20: Unused import: Callable.

The Callable type from collections.abc is imported but not used in this file.

-from collections.abc import Callable

---

488-490: API key may not match selected provider.

The API key is loaded from ANTHROPIC_API_KEY or OPENAI_API_KEY without checking if it matches the provider parameter. If user sets provider="openai" but only ANTHROPIC_API_KEY exists, the wrong key will be used.

💡 Suggested improvement

-        self.api_key = (
-            api_key or os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY")
-        )
+        if api_key:
+            self.api_key = api_key
+        elif self.provider == "claude":
+            self.api_key = os.environ.get("ANTHROPIC_API_KEY")
+        elif self.provider == "openai":
+            self.api_key = os.environ.get("OPENAI_API_KEY")
+        else:
+            self.api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY")

---

1705-1711: Hardcoded timeout values should be configurable.

The 120-second timeout is hardcoded in multiple places (lines 1711, 1777, and also in execute_fix_commands at line 1555). Consider making this a class constant or constructor parameter for configurability.

 class DiagnosisEngine:
     MAX_FIX_ATTEMPTS = 5
     MAX_STACK_DEPTH = 10
+    DEFAULT_TIMEOUT = 120

     # Then use self.DEFAULT_TIMEOUT instead of hardcoded 120

Also applies to: 1771-1777

---

1802-1823: Missing error handling for LLM API calls.

The _call_llm method doesn't handle API-specific errors like rate limits, context length exceeded, or network timeouts. While the caller has a broad try/except, specific handling would improve reliability and user feedback.

💡 Suggested error handling

     def _call_llm(self, system_prompt: str, user_prompt: str) -> str:
         """Call the LLM and return response text."""
+        if not self.client:
+            raise ValueError("LLM client not initialized")
+
         if self.provider == "claude":
-            response = self.client.messages.create(
-                model=self.model,
-                max_tokens=2048,
-                system=system_prompt,
-                messages=[{"role": "user", "content": user_prompt}],
-            )
-            return response.content[0].text
+            try:
+                response = self.client.messages.create(
+                    model=self.model,
+                    max_tokens=2048,
+                    system=system_prompt,
+                    messages=[{"role": "user", "content": user_prompt}],
+                )
+                return response.content[0].text
+            except Exception as e:
+                if "rate" in str(e).lower():
+                    raise RuntimeError(f"Rate limit exceeded: {e}") from e
+                raise

---

1893-1899: Redundant API key loading in factory function.

The get_diagnosis_engine function loads API keys from environment variables, but DiagnosisEngine.__init__ already does this when api_key=None. The factory function can be simplified.

 def get_diagnosis_engine(
     provider: str = "claude",
     debug: bool = False,
 ) -> DiagnosisEngine:
     """Factory function to create a DiagnosisEngine."""
-    api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("OPENAI_API_KEY")
-    return DiagnosisEngine(api_key=api_key, provider=provider, debug=debug)
+    return DiagnosisEngine(provider=provider, debug=debug)

---

1906-1908: Unused import in test block.

import sys on line 1907 is not used in the test block.

 if __name__ == "__main__":
-    import sys
-
     console.print("[bold]Diagnosis Engine Test[/bold]\n")

cortex/do_runner/Untitled (1)

1-1: Remove accidental placeholder file.

This single-character file looks like an accidental commit and can confuse tooling. Please delete it rather than keeping an empty placeholder.

🧹 Proposed fix

-i

cortex/system_info_generator.py (1)

519-533: Ensure httpx is declared as a dependency for Ollama support.

Line 519 imports httpx at runtime; if it isn’t declared in the project dependencies, provider="ollama" will fail with ImportError. Please confirm it’s included in the dependency manifest(s).

#!/bin/bash
# Check whether httpx is declared in dependency manifests.
rg -n "httpx" -g 'pyproject.toml' -g 'requirements*.txt' -g 'setup.cfg' -g 'setup.py'

cortex/watch_service.py (2)

68-73: Missing type hints for the log method.

Per coding guidelines, type hints are required for Python code.

Proposed fix

-    def log(self, message: str):
+    def log(self, message: str) -> None:
         """Log a message to the service log."""

---

184-317: Consider adding timeout to subprocess call in _monitor_bash_history.

The subprocess.run call at line 152-154 has a 2-second timeout, which is good. However, the inotify file descriptor is not closed in the except block if an exception occurs during the main loop.

Proposed fix for inotify cleanup

         while self.running:
             # Wait for inotify event with timeout
             r, _, _ = select.select([fd], [], [], 1.0)
             if not r:
                 continue
 
             data = os.read(fd, 4096)
             # ... processing
 
-        os.close(fd)
-        return
+        try:
+            os.close(fd)
+        except OSError:
+            pass
+        return
 
     except Exception as e:
         self.log(f"Inotify not available, using polling: {e}")
+        try:
+            os.close(fd)
+        except (OSError, NameError):
+            pass

scripts/setup_ask_do.sh (4)

20-20: Consider adding set -u for stricter variable handling.

Using set -eu (or set -euo pipefail) would catch undefined variable usage, which is a shell scripting best practice.

Proposed fix

-set -e
+set -euo pipefail

---

121-127: sed -i with pattern range may leave orphaned content if markers are malformed.

The sed command sed -i '/# Cortex Terminal Watch Hook/,/^$/d' deletes from the marker to the next empty line. If there's no empty line after the block, it could delete more than intended. Consider using a more precise end marker.

---

172-178: Sourcing venv/bin/activate may fail in subshell context.

The source venv/bin/activate inside the script activates the venv for the script's process, but this won't persist after the script ends. The warning message is helpful, but consider clarifying this to users.

---

334-356: Duplicate code between watch_hook.sh and .bashrc injection.

The same hook code is written to ~/.cortex/watch_hook.sh (lines 303-325) and also appended directly to .bashrc (lines 337-355). This creates maintenance burden. Consider sourcing the hook file from .bashrc instead.

Proposed simplification

 # Add to .bashrc
 MARKER="# Cortex Terminal Watch Hook"
 if [ -f ~/.bashrc ]; then
     if ! grep -q "$MARKER" ~/.bashrc; then
         print_step "Adding hook to .bashrc..."
         cat >> ~/.bashrc << 'EOF'

 # Cortex Terminal Watch Hook
-__cortex_last_histnum=""
-__cortex_log_cmd() {
-    # ... duplicate code ...
-}
-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+if [ -f ~/.cortex/watch_hook.sh ]; then
+    source ~/.cortex/watch_hook.sh
+fi

 alias cw="source ~/.cortex/watch_hook.sh"
 EOF

cortex/do_runner/models.py (1)

267-269: Command truncation may cut mid-word or mid-escape sequence.

node.command[:50] truncates without checking if it cuts mid-word or mid-escape. While minor, it could display oddly.

cortex/do_runner/database.py (3)

125-127: SQL injection risk in schema migration via string interpolation.

The column name and type are interpolated directly into the SQL string. While currently controlled by internal constants, this pattern is fragile.

Proposed fix using parameterized approach

         for col_name, col_type in new_columns:
             if col_name not in existing_columns:
                 try:
-                    conn.execute(f"ALTER TABLE do_runs ADD COLUMN {col_name} {col_type}")
+                    # Column names can't be parameterized, but validate against allowlist
+                    if col_name.isidentifier() and col_type.replace(" ", "").replace("_", "").isalnum():
+                        conn.execute(f"ALTER TABLE do_runs ADD COLUMN {col_name} {col_type}")
                 except sqlite3.OperationalError:
                     pass

Since the values come from hardcoded constants in this file, this is low risk but worth noting for future maintenance.

---

404-416: Weak session ID generation using MD5.

The session ID uses MD5 for hashing, which is cryptographically weak. While this is for session identification (not security), using hashlib.sha256 or secrets.token_hex would be more appropriate.

Proposed fix

+import secrets
+
     def create_session(self) -> str:
         """Create a new session and return the session ID."""
-        session_id = f"session_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}_{hashlib.md5(str(datetime.datetime.now().timestamp()).encode()).hexdigest()[:8]}"
+        session_id = f"session_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}_{secrets.token_hex(4)}"

---

276-298: Missing error handling for malformed full_data JSON.

If full_data is corrupted or missing required keys, this will raise an exception. Consider defensive handling.

Proposed fix

     def get_run(self, run_id: str) -> DoRun | None:
         """Get a specific run by ID."""
         with sqlite3.connect(str(self.db_path)) as conn:
             conn.row_factory = sqlite3.Row
             cursor = conn.execute("SELECT * FROM do_runs WHERE run_id = ?", (run_id,))
             row = cursor.fetchone()

             if row:
-                full_data = json.loads(row["full_data"])
+                try:
+                    full_data = json.loads(row["full_data"])
+                except (json.JSONDecodeError, TypeError):
+                    return None
                 run = DoRun(

cortex/do_runner/managers.py (1)

104-116: Prefix matching could allow bypass via path manipulation.

The is_protected check at line 113 uses path.startswith(protected + "/") which could be bypassed with symlinks or path traversal. Consider resolving symlinks.

Proposed enhancement

     def is_protected(self, path: str) -> bool:
         """Check if a path requires authentication for access."""
-        path = os.path.abspath(path)
+        try:
+            path = os.path.realpath(path)  # Resolve symlinks
+        except OSError:
+            path = os.path.abspath(path)
         all_protected = self.SYSTEM_PROTECTED_PATHS | self.USER_PROTECTED_PATHS

cortex/demo.py (1)

27-34: Add explicit -> None return types for helper functions.

_wait_for_enter and _section are missing return annotations. The project requires type hints for all Python code, so both should declare -> None. As per coding guidelines, please add explicit return type annotations.

✏️ Suggested fix

-def _wait_for_enter():
+def _wait_for_enter() -> None:
     """Wait for user to press enter."""
     console.print("\n[dim]Press Enter to continue...[/dim]")
     input()
 
-def _section(title: str, problem: str):
+def _section(title: str, problem: str) -> None:
     """Display a compact section header."""
     console.print(f"\n[bold cyan]{'─' * 50}[/bold cyan]")
     console.print(f"[bold white]{title}[/bold white]")
     console.print(f"[dim]{problem}[/dim]\n")

cortex/system_info_generator.py (1)

519-534: Declare httpx as a runtime dependency for Ollama.

The Ollama branch imports httpx, which will raise ImportError if it isn't listed in the runtime dependencies. Please add it to the dependency manifest (or provide a clear install-time error).

#!/bin/bash
# Verify whether httpx is declared in dependency manifests
rg -n "httpx" -g "pyproject.toml" -g "requirements*.txt" -g "setup.cfg" -g "setup.py"

cortex/watch_service.py (2)

397-405: Guard _recent_commands updates with a lock.

_is_duplicate mutates _recent_commands from multiple threads; list operations are not atomic. Wrap the membership check/append/pop in a lock to prevent races.

✏️ Suggested fix

 class CortexWatchDaemon:
     def __init__(self):
         # ...
         self._recent_commands: list[str] = []  # Last 100 commands for dedup
+        self._recent_commands_lock = threading.Lock()
 
     def _is_duplicate(self, command: str) -> bool:
         """Check if command was recently logged to avoid duplicates."""
-        if command in self._recent_commands:
-            return True
-
-        # Keep last 100 commands
-        self._recent_commands.append(command)
-        if len(self._recent_commands) > 100:
-            self._recent_commands.pop(0)
-
-        return False
+        with self._recent_commands_lock:
+            if command in self._recent_commands:
+                return True
+            self._recent_commands.append(command)
+            if len(self._recent_commands) > 100:
+                self._recent_commands.pop(0)
+            return False

---

620-624: Avoid empty username in loginctl enable-linger.

If USER is unset, an empty string is passed to loginctl, which can fail silently. Resolve a real username and only call when non-empty.

✏️ Suggested fix

+import getpass
@@
-        subprocess.run(["loginctl", "enable-linger", os.getenv("USER", "")], capture_output=True)
+        user = os.getenv("USER") or getpass.getuser()
+        if user:
+            subprocess.run(["loginctl", "enable-linger", user], capture_output=True)

cortex/do_runner/diagnosis_v2.py (4)

1318-1325: Sanitize cmd_name in apt-file search or avoid shell=True.

cmd_name comes from the original command and is interpolated into a shell string, creating injection risk. Validate allowed characters and run with shell=False, or use shutil.which before falling back to apt‑file.

✏️ Suggested fix

-                result = subprocess.run(
-                    f"apt-file search --regexp 'bin/{cmd_name}$' 2>/dev/null | head -1 | cut -d: -f1",
-                    shell=True,
-                    capture_output=True,
-                    text=True,
-                    timeout=10,
-                )
+                if not re.fullmatch(r"[A-Za-z0-9._+-]+", cmd_name):
+                    return None
+                result = subprocess.run(
+                    ["apt-file", "search", "--regexp", f"bin/{cmd_name}$"],
+                    capture_output=True,
+                    text=True,
+                    timeout=10,
+                )
+                if result.returncode == 0 and result.stdout.strip():
+                    return result.stdout.splitlines()[0].split(":", 1)[0].strip()

---

1453-1459: Avoid enabling plaintext git credential storage without opt‑in.

git config --global credential.helper store writes credentials in plain text to ~/.git-credentials. Use a safer helper (cache/manager) or prompt the user to opt in with a clear warning.

✏️ Suggested fix

-                subprocess.run("git config --global credential.helper store", shell=True)
+                console.print(
+                    "[yellow]⚠ Git credential.helper 'store' saves passwords in plain text.[/yellow]"
+                )
+                subprocess.run("git config --global credential.helper cache", shell=True)

---

1463-1471: Do not embed credentials in pip index URL.

pip config set global.index-url https://{username}:{password}@... persists secrets to config and shell history. Set the index URL without credentials and supply auth via env (PIP_INDEX_URL) or keyring/netrc.

✏️ Suggested fix

-                login_cmd = f"pip config set global.index-url https://{username}:{{password}}@pypi.org/simple/"
+                login_cmd = "pip config set global.index-url https://pypi.org/simple/"
+                # Provide credentials via env (PIP_INDEX_URL) or a secure helper instead

---

1549-1556: Avoid shell=True when executing fix commands.

Resolved variables can be user/LLM-derived; executing with shell=True invites injection. Prefer shlex.split + shell=False and only fall back to shell=True for truly complex commands.

✏️ Suggested fix

+import shlex
@@
-                result = subprocess.run(
-                    command,
-                    shell=True,
-                    capture_output=True,
-                    text=True,
-                    timeout=120,
-                )
+                try:
+                    args = shlex.split(command)
+                    result = subprocess.run(
+                        args,
+                        capture_output=True,
+                        text=True,
+                        timeout=120,
+                    )
+                except ValueError:
+                    result = subprocess.run(
+                        command,
+                        shell=True,
+                        capture_output=True,
+                        text=True,
+                        timeout=120,
+                    )

cortex/do_runner/executor.py (2)

151-205: Auto-repair can execute sudo‑required fixes without explicit approval.

Line 151 only checks protected paths, so repairs like apt install/systemctl can run without confirmation when confirm_callback is missing or new_paths is empty; _execute_command will still prepend sudo. Please gate on _needs_sudo() or explicit sudo too, and skip auto‑fix when confirmation isn’t available. As per coding guidelines, avoid silent sudo execution.

🔧 Suggested guard

-                new_paths = self._identify_paths_needing_privileges(fix_commands)
-                if new_paths and confirm_callback:
+                new_paths = self._identify_paths_needing_privileges(fix_commands)
+                needs_privilege = new_paths or any(
+                    self._needs_sudo(cmd) or cmd.strip().startswith("sudo")
+                    for cmd in fix_commands
+                )
+                if needs_privilege and confirm_callback:

---

66-112: Add test coverage and fix silent sudo execution in TaskTreeExecutor.

This execution engine requires >80% test coverage per coding guidelines (currently no tests exist). Additionally, the file matches the **/*exec*.py pattern requiring explicit user approval for elevated permissions. However, _execute_command() silently adds sudo to qualifying commands without checking confirm_callback first—only repair tasks with privilege-needing paths verify approval. Refactor to require confirm_callback approval before any sudo elevation, then add comprehensive unit tests for success, failure, auto-repair, permission gating, and sudo approval flows.

cortex/do_runner/models.py (2)

90-96: get_depth() never advances—this can infinite loop.

Line 94 assigns node = node, so the loop never progresses when parent_id is set. This can hang any caller. Consider passing a TaskTree to resolve parents or store a direct parent reference.

🧭 Example fix

-    def get_depth(self) -> int:
+    def get_depth(self, tree: "TaskTree") -> int:
         """Get the depth of this node in the tree."""
         depth = 0
-        node = self
-        while node.parent_id:
-            depth += 1
-            node = node
+        current_id = self.parent_id
+        while current_id:
+            depth += 1
+            parent = tree.get_task(current_id)
+            current_id = parent.parent_id if parent else None
         return depth

---

349-356: Output preview always shows “...” even when not truncated.

Line 356 appends ellipsis unconditionally. This makes short outputs look truncated.

✅ Fix

-            if cmd.output:
-                lines.append(f"  Output: {cmd.output[:500]}...")
+            if cmd.output:
+                preview = cmd.output[:500]
+                suffix = "..." if len(cmd.output) > 500 else ""
+                lines.append(f"  Output: {preview}{suffix}")

scripts/setup_ask_do.py (1)

289-347: Don’t clobber existing PROMPT_COMMAND in hooks.

Lines 310 and 346 overwrite PROMPT_COMMAND, which can break users’ prompt tooling. Please append instead of replace.

💡 Safer composition

-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+export PROMPT_COMMAND="${PROMPT_COMMAND:+$PROMPT_COMMAND; }history -a; __cortex_log_cmd"

cortex/do_runner/terminal.py (2)

714-739: Preserve existing PROMPT_COMMAND when installing hooks.

Line 738 overwrites PROMPT_COMMAND, which can break user prompt tooling. Compose with the existing value.

💡 Safer composition

-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+export PROMPT_COMMAND="${PROMPT_COMMAND:+$PROMPT_COMMAND; }history -a; __cortex_log_cmd"

---

1489-1512: Never auto‑execute sudo, even with NOPASSWD.

Lines 1492–1506 attempt a sudo -n true probe and then proceed to run sudo commands without interactive approval. This violates the “no automatic sudo execution” guardrail. As per coding guidelines, always require explicit user approval for sudo.

✅ Remove auto‑sudo probe

-            if needs_sudo:
-                try:
-                    check_sudo = subprocess.run(
-                        ["sudo", "-n", "true"], capture_output=True, timeout=5
-                    )
-
-                    if check_sudo.returncode != 0:
-                        sudo_commands_pending.append(fix_cmd)
-                        results.append((fix_cmd, "sudo", None))
-                        console.print(
-                            f"  [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]"
-                        )
-                        continue
-                except Exception:
-                    sudo_commands_pending.append(fix_cmd)
-                    results.append((fix_cmd, "sudo", None))
-                    console.print(
-                        f"  [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]"
-                    )
-                    continue
+            if needs_sudo:
+                sudo_commands_pending.append(fix_cmd)
+                results.append((fix_cmd, "sudo", None))
+                console.print(
+                    f"  [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]"
+                )
+                continue

cortex/watch_service.py (1)

467-545: Add return type annotations for daemon lifecycle APIs.

start, stop, and status (plus main/log) lack explicit return types. Please add return annotations to satisfy the typing requirement. As per coding guidelines, add explicit return type annotations.

✏️ Suggested fix

-    def start(self):
+    def start(self) -> bool:
@@
-    def stop(self):
+    def stop(self) -> tuple[bool, str]:
@@
-    def status(self) -> dict:
+    def status(self) -> dict[str, Any]:
@@
-def main():
+def main() -> None:

cortex/do_runner/database.py (1)

26-41: Add return type annotations for internal DB helpers.

_ensure_directory, _init_db, _migrate_schema, update_session, and end_session lack return annotations. Please add -> None where appropriate. As per coding guidelines, add explicit return type annotations.

✏️ Suggested fix

-    def _ensure_directory(self):
+    def _ensure_directory(self) -> None:
@@
-    def _init_db(self):
+    def _init_db(self) -> None:
@@
-    def _migrate_schema(self, conn: sqlite3.Connection):
+    def _migrate_schema(self, conn: sqlite3.Connection) -> None:
@@
-    def update_session(
-        self, session_id: str, query: str | None = None, increment_runs: bool = False
-    ):
+    def update_session(
+        self, session_id: str, query: str | None = None, increment_runs: bool = False
+    ) -> None:
@@
-    def end_session(self, session_id: str):
+    def end_session(self, session_id: str) -> None:

scripts/setup_ask_do.py (1)

43-67: Add return type hints + docstring for public APIs to meet guidelines.

Public helpers (e.g., print_header, print_step, print_success, main) are missing return type annotations, and main() lacks a docstring. As per coding guidelines, public APIs should be fully typed and documented.

✍️ Example updates

-def print_header(text: str):
+def print_header(text: str) -> None:
     """Print a section header."""
@@
-def print_step(text: str):
+def print_step(text: str) -> None:
@@
-def print_success(text: str):
+def print_success(text: str) -> None:
@@
-def print_warning(text: str):
+def print_warning(text: str) -> None:
@@
-def print_error(text: str):
+def print_error(text: str) -> None:
@@
-def main():
+def main() -> int:
+    """Run the ask --do setup wizard."""

Also applies to: 544-605

cortex/watch_service.py (2)

397-405: Protect _recent_commands with a lock.
Concurrent membership check + append + pop across threads is non-atomic and can race.

---

623-624: Avoid passing an empty username to loginctl enable-linger.
If USER is unset, Line 624 passes an empty string. Resolve a real username (getpass/pwd) and guard the call.

🔧 Suggested fix

-        subprocess.run(["loginctl", "enable-linger", os.getenv("USER", "")], capture_output=True)
+        import getpass
+
+        user = os.getenv("USER") or getpass.getuser()
+        if user:
+            subprocess.run(["loginctl", "enable-linger", user], capture_output=True)

cortex/demo.py (1)

27-34: Add -> None return types for helper functions.

As per coding guidelines, these helpers should have explicit return type annotations.

✏️ Proposed fix

-def _wait_for_enter():
+def _wait_for_enter() -> None:
     """Wait for user to press enter."""
     console.print("\n[dim]Press Enter to continue...[/dim]")
     input()

-def _section(title: str, problem: str):
+def _section(title: str, problem: str) -> None:
     """Display a compact section header."""
     console.print(f"\n[bold cyan]{'─' * 50}[/bold cyan]")
     console.print(f"[bold white]{title}[/bold white]")
     console.print(f"[dim]{problem}[/dim]\n")

cortex/do_runner/managers.py (1)

220-237: chmod fallback grants world-level access.

The fallback uses o+rw, which makes the file world-readable/writable and exceeds the intended privilege scope.

cortex/do_runner/models.py (2)

90-97: Infinite loop in get_depth().

node = node never advances, so this loops forever when parent_id is set. It needs a parent reference or a tree lookup.

---

349-357: Output always marked as truncated.

cmd.output[:500]}... appends ... even when the output is shorter than 500 chars.

cortex/do_runner/diagnosis_v2.py (4)

1318-1326: Avoid shell=True with user-influenced cmd_name.

apt-file search is built via a shell string using cmd_name, which can be manipulated.

---

1458-1459: credential.helper store saves passwords in plaintext.

This writes credentials to ~/.git-credentials in plain text; please warn and/or require explicit opt-in or use a safer helper.

---

1463-1471: Credentials embedded in pip index URL.

This stores secrets in config and may expose them in logs/history.

---

1549-1556: Shell execution with substituted variables risks injection.

execute_fix_commands runs shell=True after variable substitution; use shlex.split and shell=False, or validate/escape inputs.

cortex/do_runner/executor.py (1)

151-208: Privileged auto-repairs can run without explicit approval.

The gate only checks protected paths; sudo-required repairs (e.g., apt, systemctl) can execute automatically, and when confirm_callback is None even protected-path repairs run. This violates the explicit-approval requirement for privileged commands. As per coding guidelines, require confirmation for any sudo-needed repair, and skip/defer if confirmation isn't available.

🔧 Suggested guard

-                new_paths = self._identify_paths_needing_privileges(fix_commands)
-                if new_paths and confirm_callback:
+                new_paths = self._identify_paths_needing_privileges(fix_commands)
+                needs_privilege = new_paths or any(
+                    self._needs_sudo(cmd) or cmd.strip().startswith("sudo")
+                    for cmd in fix_commands
+                )
+                if needs_privilege:
+                    if not confirm_callback:
+                        console.print("[yellow]⚠ Privileged repairs require approval; skipping auto-repair.[/yellow]")
+                        task.status = CommandStatus.FAILED
+                        task.reasoning = "Privileged repair requires explicit approval"
+                        return False, was_repaired
+
+                if needs_privilege and confirm_callback:

cortex/system_info_generator.py (2)

388-400: Align prompt with CommandValidator (no chaining).

The prompt still allows || fallbacks while validation blocks chaining, so generated commands will be rejected. Keep the prompt consistent with validator behavior.

🔧 Suggested prompt adjustment

-- NEVER use dangerous command chaining (;, &&, ||) except for fallback patterns
-- Commands should handle missing files/tools gracefully using || echo fallbacks
+- NEVER use command chaining (;, &&, ||)
+- Commands should handle missing files/tools gracefully without chaining

---

518-534: Declare httpx as a dependency for the Ollama path.

httpx is imported for the Ollama client, but it isn’t declared in dependency manifests. This will raise ImportError when provider="ollama" is used.

cortex/do_runner/verification.py (1)

19-33: Avoid double‑sudo in _execute_command.

When needs_sudo=True, the code prepends sudo and then wraps with ["sudo","bash","-c", ...], yielding sudo sudo ... and breaking execution.

✅ Safer execution pattern

-            if needs_sudo and not cmd.strip().startswith("sudo"):
-                cmd = f"sudo {cmd}"
-
-            result = subprocess.run(
-                ["sudo", "bash", "-c", cmd] if needs_sudo else cmd,
-                shell=not needs_sudo,
+            exec_cmd = ["sudo", "bash", "-c", cmd] if needs_sudo else cmd
+            result = subprocess.run(
+                exec_cmd,
+                shell=not needs_sudo,
                 capture_output=True,
                 text=True,
                 timeout=timeout,
             )

Also applies to: 817-831, 997-1011

cortex/do_runner/terminal.py (2)

714-739: Preserve existing PROMPT_COMMAND when installing the hook.

Current setup overwrites any existing PROMPT_COMMAND, breaking user prompt tooling and other hooks.

💡 Safer PROMPT_COMMAND composition

-export PROMPT_COMMAND='history -a; __cortex_log_cmd'
+export PROMPT_COMMAND="${PROMPT_COMMAND:+$PROMPT_COMMAND; }history -a; __cortex_log_cmd"

---

1489-1512: Never auto‑execute sudo, even with NOPASSWD.

The NOPASSWD probe auto‑runs sudo commands, which violates the “no automatic sudo execution” guardrail. Sudo fixes should always be queued for manual execution.

✅ Safer behavior

-            if needs_sudo:
-                try:
-                    check_sudo = subprocess.run(
-                        ["sudo", "-n", "true"], capture_output=True, timeout=5
-                    )
-
-                    if check_sudo.returncode != 0:
-                        sudo_commands_pending.append(fix_cmd)
-                        results.append((fix_cmd, "sudo", None))
-                        console.print(
-                            f"  [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]"
-                        )
-                        continue
-                except Exception:
-                    sudo_commands_pending.append(fix_cmd)
-                    results.append((fix_cmd, "sudo", None))
-                    console.print(
-                        f"  [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]"
-                    )
-                    continue
+            if needs_sudo:
+                sudo_commands_pending.append(fix_cmd)
+                results.append((fix_cmd, "sudo", None))
+                console.print(
+                    f"  [dim][{i}/{len(actionable)}][/dim] [yellow]![/yellow] {fix_cmd[:55]}... [dim](needs sudo)[/dim]"
+                )
+                continue

cortex/cli.py (1)

855-861: Allow question to be optional in do-mode.
Line 855 types question as str, but callers pass None when --do is used without an initial question (Line 5581-5585). Update the annotation/docstring to str | None to keep typing accurate.

♻️ Suggested tweak

-    def ask(self, question: str, do_mode: bool = False) -> int:
+    def ask(self, question: str | None, do_mode: bool = False) -> int:

As per coding guidelines, type hints should reflect actual usage.

cortex/do_runner/managers.py (1)

58-88: Add explicit return type annotations for helpers.

As per coding guidelines, type hints are required for Python code. Please add -> None on helper methods like __init__, _ensure_config_dir, _load_user_paths, and _save_user_paths.

✏️ Example fix

-    def __init__(self):
+    def __init__(self) -> None:
         self.config_file = Path.home() / ".cortex" / "protected_paths.json"
         self._ensure_config_dir()
         self._load_user_paths()

-    def _ensure_config_dir(self):
+    def _ensure_config_dir(self) -> None:
         """Ensure the config directory exists."""
         try:
             self.config_file.parent.mkdir(parents=True, exist_ok=True)
         except OSError:
             self.config_file = Path("/tmp") / ".cortex" / "protected_paths.json"
             self.config_file.parent.mkdir(parents=True, exist_ok=True)

-    def _load_user_paths(self):
+    def _load_user_paths(self) -> None:
         """Load user-configured protected paths."""
         if self.config_file.exists():
             try:
                 with open(self.config_file) as f:
                     data = json.load(f)
                     self.USER_PROTECTED_PATHS = set(data.get("paths", []))
             except (json.JSONDecodeError, OSError):
                 pass

-    def _save_user_paths(self):
+    def _save_user_paths(self) -> None:
         """Save user-configured protected paths."""
         try:
             self.config_file.parent.mkdir(parents=True, exist_ok=True)
             with open(self.config_file, "w") as f:
                 json.dump({"paths": list(self.USER_PROTECTED_PATHS)}, f, indent=2)
         except OSError as e:
             console.print(f"[yellow]Warning: Could not save protected paths: {e}[/yellow]")

cortex/do_runner/models.py (2)

85-88: Add return type for add_child.

As per coding guidelines, add an explicit return type (e.g., -> None).

✏️ Minimal fix

-    def add_child(self, child: "TaskNode"):
+    def add_child(self, child: "TaskNode") -> None:
         """Add a child task."""
         child.parent_id = self.id
         self.children.append(child)

---

238-278: Add return types for print_tree / _print_node.

As per coding guidelines, annotate these helpers with -> None for clarity and consistency.

✏️ Example

-    def print_tree(self, indent: str = ""):
+    def print_tree(self, indent: str = "") -> None:
         """Print the task tree structure."""
         for i, root in enumerate(self.root_tasks):
             is_last = i == len(self.root_tasks) - 1
             self._print_node(root, indent, is_last)

-    def _print_node(self, node: TaskNode, indent: str, is_last: bool):
+    def _print_node(self, node: TaskNode, indent: str, is_last: bool) -> None:
         """Print a single node with its children."""

cortex/do_runner/database.py (1)

205-206: Add return type annotations for public DB APIs.

As per coding guidelines, public methods should include explicit return types (e.g., save_run(...) -> str), and similarly for other public methods in this class.

✏️ Example fix

-    def save_run(self, run: DoRun) -> str:
+    def save_run(self, run: DoRun) -> str:
         """Save a do run to the database with detailed command information."""

(Please apply the same pattern to the other public methods.)

cortex/system_info_generator.py (1)

574-576: Fix history typing to match actual values.

history is annotated as dict[str, str], but you store booleans and possibly non‑string values. Use dict[str, Any] or a TypedDict to keep type hints accurate. As per coding guidelines, keep type annotations precise for new code.

♻️ Suggested type fix

-        history: list[dict[str, str]] = []
+        history: list[dict[str, Any]] = []